package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo3Parallelism {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * 设置并行度
     * 1、在代码中统一设置   -- 优先级比提交时设置高
     * 2、在提交任务是设置并行度  命令行-p
     * 3、每一个算子可以单独设置并行度  - 优先级最高
     *
     * 在项目上一般使用第二种方式，可以灵活修改并行度不需要重新打包项目
     *
     * flink资源申请有最高的并行度决定，一个并行度占用一个资源
     * 并行度由数据量决定，并行度越高每秒能处理的数据量越多
     *
     *
     * flink task合并问题
     * 当上下游并行度一致同时中间没有shuffle时可以合并成同一个task
     *
     */

    //env.setParallelism(2)

    //flink从socket中读取数据并行度只能是1
    val linesDS: DataStream[String] = env
      .socketTextStream("master", 8888)

    val wordsDS: DataStream[String] = linesDS
      .flatMap(_.split(","))
      //设置算子的并行度
      .setParallelism(2)
      //设置算子的名称可以在网页中识别
      .name("拆分数据")

    val kvDS: DataStream[(String, Int)] = wordsDS
      .map((_, 1))
      .setParallelism(3)
      .name("转换成kv")

    //keyBy不能设置并行度
    val keyByDS: KeyedStream[(String, Int), String] = kvDS
      .keyBy(_._1)

    val countDS: DataStream[(String, Int)] = keyByDS
      .sum(1)
      .setParallelism(2)
      .name("求和")

    countDS
      .print()
      .setParallelism(1)
      .name("打印结果")

    env.execute()
  }

}
